package test.threads.queue.stressed;

import java.util.concurrent.*;

import com.google.common.base.Throwables;
import com.lmax.disruptor.EventFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import test.helpers.Config;
import test.threads.queue.common.IUsefulWorkEmulator;

/**
 * (source) -> 0 -> 1 -> 2 -> ... -> N
 *
 * @author cheremin
 * @since 27.02.12,  15:00
 */
public class ABQxNTask<E> extends AbstractTask<E> {
    private static final Log log = LogFactory.getLog( ABQxNTask.class );

    private final int nodes;

    public ABQxNTask( final Config config ) {
        super( config );
        nodes = config.getAsInt( "pipeline.nodes" );
    }


    @Override
    public int workersRequired() {
        return nodes;
    }

    private IUsefulWorkEmulator<E> lastEmulator;
    private BlockingQueue<E> outbox;
    private PassThroughNode<E>[] tasks;

    @Override
    protected EventEnqueuer _initialize( final Executor threadPool,
                                         final IUsefulWorkEmulator<E>[] emulators,
                                         final EventFactory<E> eventFactory ) {
        final BlockingQueue<E> inbox = createQueue();


        BlockingQueue<E> _outbox = inbox;
        tasks = new PassThroughNode[nodes - 1];
        for ( int i = 0; i < tasks.length; i++ ) {
            final BlockingQueue<E> _inbox = _outbox;
            _outbox = createQueue();

            final String name = String.valueOf( i );
            final PassThroughNode<E> task = new PassThroughNode<E>(
                    name,
                    _inbox,
                    emulators[i],
                    _outbox
            );

            tasks[i] = task;
            threadPool.execute( task );
        }

        //this is last task data
        this.outbox = _outbox;
        lastEmulator = emulators[nodes - 1];

        return new Enqueuer( eventsInBatch(), eventFactory, inbox );
    }

    @Override
    public void terminate() {
        super.terminate();
        for ( final PassThroughNode<E> task : tasks ) {
            task.terminate();
        }
    }

    private ArrayBlockingQueue<E> createQueue() {
        return new ArrayBlockingQueue<E>( bufferSize() );
    }

    @Override
    protected long dequeueBatchEvents( final int eventsInBatch ) throws Exception {
        for ( int i = 0; i < eventsInBatch; i++ ) {
            final E event = outbox.take();
            lastEmulator.spendCPUTime( event );
        }

        return 0;
    }

    private static class PassThroughNode<E> implements Runnable {
        protected final String name;
        protected final IUsefulWorkEmulator<E> emulator;
        protected final BlockingQueue<E> inbox;
        protected final BlockingQueue<E> outbox;

        public PassThroughNode( final String name,
                                final BlockingQueue<E> inbox,
                                final IUsefulWorkEmulator<E> emulator,
                                final BlockingQueue<E> outbox ) {
            this.name = name;
            this.emulator = emulator;
            this.inbox = inbox;
            this.outbox = outbox;
        }

        private volatile Thread currentThread;

        @Override
        public void run() {
            try {
                currentThread = Thread.currentThread();
                while ( true ) {
                    makeStep();
                }
            } catch ( InterruptedException e ) {
                log.info( "PassThroughNode[" + name + "] interrupted" );
            } catch ( Throwable t ) {
                log.error( "PassThroughNode[" + name + "] exited unexpectedly", t );
                throw Throwables.propagate( t );
            }
        }

        protected void makeStep() throws InterruptedException {
            final E taken = inbox.take();
            emulator.spendCPUTime( taken );
            outbox.put( taken );
        }

        public void terminate() {
            currentThread.interrupt();
        }

        @Override
        public String toString() {
            return name;
        }
    }

    private class Enqueuer extends EventEnqueuer<E> {
        private final BlockingQueue<E> inbox;

        private Enqueuer( final int eventsInBatch,
                          final EventFactory<E> factory,
                          final BlockingQueue<E> inbox ) {
            super( eventsInBatch, factory );
            this.inbox = inbox;
        }

        @Override
        protected void enqueueEventsBatch( final int eventsInBatch,
                                           final EventFactory<E> factory ) throws Exception {
            for ( int i = 0; i < eventsInBatch; i++ ) {
                inbox.put( factory.newInstance() );
            }
        }
    }
}
